package com.ziyucoding.springboot.labs.lab2.kafkademo.producer;

import com.ziyucoding.springboot.labs.lab2.kafkademo.message.DemoMessage;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;

import javax.annotation.Resource;
import java.util.concurrent.ExecutionException;

/**
 * 生产者示例
 *
 * @author ziyucoding
 * @date 2022-10-11
 **/
@Component
public class DemoProducer {

    @Resource
    private KafkaTemplate<Object, Object> kafkaTemplate;

    /**
     * 同步发送消息
     *
     * @param id
     * @return org.springframework.kafka.support.SendResult
     **/
    public SendResult syncSend(Integer id) throws ExecutionException, InterruptedException {
        // 创建消息
        DemoMessage message = new DemoMessage();
        message.setId(id);
        // 同步发送消息
        SendResult result = kafkaTemplate.send(DemoMessage.TOPIC, message).get();
        return result;
    }

    /**
     * 异步发送消息
     *
     * @param id
     * @return org.springframework.util.concurrent.ListenableFuture<org.springframework.kafka.support.SendResult < java.lang.Object, java.lang.Object>>
     **/
    public ListenableFuture<SendResult<Object, Object>> asyncSend(Integer id) {
        // 创建消息
        DemoMessage message = new DemoMessage();
        message.setId(id);
        // 异步发送消息
        ListenableFuture<SendResult<Object, Object>> result = kafkaTemplate.send(DemoMessage.TOPIC, message);
        return result;
    }
}
